Upgrade Kafka read/write transforms without upgrading the pipeline #29362

chamikaramj · 2023-11-09T03:57:24Z

This allows pipeline authors to upgrade Beam Kafka read/write transforms to a newer Beam version while keeping the rest of the pipeline in an older Beam version.

This will be useful for pipeline authors who wish to use newer releases of Kafka read/write transforms due to bug fixes, perf. improvements etc. but are unable to upgrade the full pipeline (for example, due to organizational restrictions).

Issue is #27943

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

github-actions · 2023-11-09T05:05:59Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

chamikaramj · 2023-11-09T18:18:18Z

assign set of reviewers

github-actions · 2023-11-09T18:20:08Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @bvolpato for label java.
R: @damccorm for label build.
R: @johnjcasey for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

johnjcasey · 2023-11-09T18:21:55Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+    "EqualsIncompatibleType",
+  })
+  public static @Nullable String findUpgradeURN(
+      org.apache.beam.sdk.transforms.PTransform transform) {


can we fix this import?

johnjcasey · 2023-11-09T18:23:09Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+              ? extends TransformPayloadTranslator>
+          entry : registrar.getTransformPayloadTranslators().entrySet()) {
+        if (entry.getKey().equals(transform.getClass())) {
+          return entry.getValue().getUrn();


Do we care if there are multiple potential matches?

The contract is to upgrade all transforms that match the URN (and TransformUpgrader already handles this).

johnjcasey · 2023-11-09T18:28:49Z

.../io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java

@@ -0,0 +1,578 @@
+/*


This class looks like it contains a bunch of relatively boilerplate code. Given that we would want translation and upgrades for all pre-built composites eventually, is there any way to try and avoid this boilerplate?

So all we do here is building a Row object from the transform object and vice versa using utilities available in the Beam Schema library. The rest of the boilerplate here is mostly error/null checks which cannot be refactored away easily. We do have fromByteArray/toByteArray methods here to convert objects that do not have a Schema defined (for example, user specified functions) but these are pretty small and different classes can choose to do the serialization in different ways.

Do you have a specific refactoring in mind to simplify this code ?

I didn't, good to know this was thought through

chamikaramj

Thanks.

chamikaramj · 2023-11-10T18:23:00Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+              ? extends TransformPayloadTranslator>
+          entry : registrar.getTransformPayloadTranslators().entrySet()) {
+        if (entry.getKey().equals(transform.getClass())) {
+          return entry.getValue().getUrn();


The contract is to upgrade all transforms that match the URN (and TransformUpgrader already handles this).

chamikaramj · 2023-11-10T21:45:06Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+    "EqualsIncompatibleType",
+  })
+  public static @Nullable String findUpgradeURN(
+      org.apache.beam.sdk.transforms.PTransform transform) {


chamikaramj · 2023-11-10T21:50:12Z

.../io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java

@@ -0,0 +1,578 @@
+/*


So all we do here is building a Row object from the transform object and vice versa using utilities available in the Beam Schema library. The rest of the boilerplate here is mostly error/null checks which cannot be refactored away easily. We do have fromByteArray/toByteArray methods here to convert objects that do not have a Schema defined (for example, user specified functions) but these are pretty small and different classes can choose to do the serialization in different ways.

Do you have a specific refactoring in mind to simplify this code ?

codecov · 2023-11-11T06:10:32Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (9b1040b) 38.32% compared to head (85288cf) 38.34%.
Report is 2 commits behind head on master.

❗ Current head 85288cf differs from pull request most recent head aff97ef. Consider uploading reports for the commit aff97ef to get more accurate results

Additional details and impacted files

@@            Coverage Diff             @@
##           master   #29362      +/-   ##
==========================================
+ Coverage   38.32%   38.34%   +0.02%     
==========================================
  Files         694      693       -1     
  Lines      102373   102227     -146     
==========================================
- Hits        39231    39200      -31     
+ Misses      61549    61436     -113     
+ Partials     1593     1591       -2

Flag	Coverage Δ
python	`29.88% <ø> (+<0.01%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

chamikaramj · 2023-11-15T18:32:21Z

PTAL

chamikaramj · 2023-11-16T20:42:43Z

Run Java PreCommit

chamikaramj · 2023-11-17T18:13:15Z

Thanks.

github-actions bot added java build io runners kafka core labels Nov 9, 2023

github-actions bot added the Next Action: Reviewers label Nov 9, 2023

johnjcasey reviewed Nov 9, 2023

View reviewed changes

chamikaramj commented Nov 10, 2023

View reviewed changes

chamikaramj force-pushed the upgrade_kafka_transforms branch from 1069662 to 85288cf Compare November 11, 2023 00:56

chamikaramj added 2 commits November 15, 2023 10:10

Upgrade Kafka read/write transforms without upgrading the pipeline

988c084

Address reviewer comments

aff97ef

chamikaramj force-pushed the upgrade_kafka_transforms branch from 85288cf to aff97ef Compare November 15, 2023 18:31

johnjcasey approved these changes Nov 17, 2023

View reviewed changes

chamikaramj merged commit 67dbdd1 into apache:master Nov 17, 2023
31 checks passed

chamikaramj mentioned this pull request Nov 28, 2023

Support for upgrading BigQuery transforms without upgrading the pipeline #29550

Merged

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Upgrade Kafka read/write transforms without upgrading the pipeline #29362

Upgrade Kafka read/write transforms without upgrading the pipeline #29362

chamikaramj commented Nov 9, 2023

github-actions bot commented Nov 9, 2023

chamikaramj commented Nov 9, 2023

github-actions bot commented Nov 9, 2023

johnjcasey Nov 9, 2023

chamikaramj Nov 10, 2023

johnjcasey Nov 9, 2023

chamikaramj Nov 10, 2023

johnjcasey Nov 9, 2023

chamikaramj Nov 10, 2023 •

edited

Loading

johnjcasey Nov 17, 2023

chamikaramj left a comment

chamikaramj Nov 10, 2023

chamikaramj Nov 10, 2023

chamikaramj Nov 10, 2023 •

edited

Loading

codecov bot commented Nov 11, 2023 •

edited

Loading

chamikaramj commented Nov 15, 2023

chamikaramj commented Nov 16, 2023

chamikaramj commented Nov 17, 2023

Upgrade Kafka read/write transforms without upgrading the pipeline #29362

Upgrade Kafka read/write transforms without upgrading the pipeline #29362

Conversation

chamikaramj commented Nov 9, 2023

GitHub Actions Tests Status (on master branch)

github-actions bot commented Nov 9, 2023

chamikaramj commented Nov 9, 2023

github-actions bot commented Nov 9, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj Nov 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj Nov 10, 2023 • edited Loading

Choose a reason for hiding this comment

codecov bot commented Nov 11, 2023 • edited Loading

Codecov Report

chamikaramj commented Nov 15, 2023

chamikaramj commented Nov 16, 2023

chamikaramj commented Nov 17, 2023

chamikaramj Nov 10, 2023 •

edited

Loading

chamikaramj Nov 10, 2023 •

edited

Loading

codecov bot commented Nov 11, 2023 •

edited

Loading